/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2022-2022. All rights reserved.
 */

import cluster from 'cluster';
import { parentPort, Worker, WorkerOptions } from 'worker_threads';

import { HookType } from '../../../../common/const/hook-const.js';
import { noop } from '../../../../common/util/noop.js';
import { WatchEvent } from '../../../common/daemon-protocol/build-event.js';
import { WORKER_PROCESS_EVENT_ID } from '../../../daemon/cluster/worker-process-event-id.js';
import { CoreTaskImpl } from '../../../external/task/core-task-impl.js';
import { HvigorLogger } from '../../../log/hvigor-log.js';
import { DurationEvent, DurationEventState } from '../../../metrics/event/duration-event.js';
import { MetricLogType } from '../../../metrics/event/log-event.js';
import { HvigorLifecycleHook } from '../../lifecycle/hook/hvigor-lifecycle-hook.js';
import { PoolConstant } from '../constant/constant.js';

const log = HvigorLogger.getLogger('daemon');
class WorkerItem {
  worker: Worker;
  hasCustomTerminate: boolean;

  constructor(worker: Worker, hasCustomTerminate: boolean) {
    this.worker = worker;
    this.hasCustomTerminate = hasCustomTerminate;
  }
}

const genPromiseAndTriggers = (task: CoreTaskImpl, startTime: number) => {
  let reject: (reason?: unknown) => void = noop;
  let resolve: (value: unknown) => void = noop;

  const promise = new Promise((_resolve, _reject) => {
    reject = (rej) => {
      task.taskExecutedStatus.setWorkerTimePeriod([startTime, Number(process.hrtime.bigint())]);
      _reject(rej);
    };
    resolve = (res) => {
      task.taskExecutedStatus.setWorkerTimePeriod([startTime, Number(process.hrtime.bigint())]);
      _resolve(res);
    };
  });

  return { reject, resolve, promise };
};

async function handleResponse(
  response: any,
  task: CoreTaskImpl,
  workerId: number,
  isWatchWorker: boolean,
  callback: Function,
  subDurationEvent?: DurationEvent,
) {
  if (response.event === PoolConstant.WORK_DONE) {
    subDurationEvent?.stop(DurationEventState.SUCCESS);
    subDurationEvent?.setLog(subDurationEvent?.getName(), MetricLogType.INFO);
    await callback(response.returnVal);
  } else if (response.event === PoolConstant.WORK_ERROR) {
    subDurationEvent?.stop(DurationEventState.FAILED);
    subDurationEvent?.setLog(subDurationEvent?.getName(), MetricLogType.ERROR);
    await callback(response.returnVal);
    const errorMsg = response.error;
    const noticeMsg = `${task.getName()} ${isWatchWorker ? 'watch' : 'normal'} work[${workerId}] failed.`;
    log.debug(noticeMsg);
    throw new Error(errorMsg || noticeMsg);
  } else {
    throw new Error(`Unknown worker event: ${response}.`);
  }
}

/**
 * watch worker和normal worker需要共享worker id分配器，否则重复的id会导致build analyzer显示错乱
 */
class WorkerCountManager {
  // cnt需要从1开始，因为0被指定为主线程的worker id
  private static cnt = 1;

  static allocateWorkerId() {
    return this.cnt++;
  }

  // 不要随便调用， 重置可能会导致后续watch任务分配到相同的id，可能会导致sessionmap中丢失线程信息，从而导致watch线程无法正常关闭。
  static reset() {
    this.cnt = 1;
  }
}

/**
 * 用于watch任务的线程管理
 *
 * @since 2022/12/7
 */
class WatchWorker {
  private readonly workerMap: Map<number, WorkerItem> = new Map<number, WorkerItem>();

  constructor() {
    this.addListenersOnSessionManager();
  }

  getWorker(id: number): Worker | undefined {
    return this.workerMap.get(id)?.worker;
  }

  sendCompileMsgToWorker() {
    this.workerMap.forEach((value, key) => {
      value.worker.postMessage({event: 'HOT_COMPILE'});
    });
  }

  createWorker(task: CoreTaskImpl, filename: string, hasCustomTerminate: boolean, callback: Function, options?: WorkerOptions): string {
    const startTime = Number(process.hrtime.bigint());
    const worker: Worker = new Worker(filename, options);
    const workerId = WorkerCountManager.allocateWorkerId();
    this.workerMap.set(workerId, new WorkerItem(worker, hasCustomTerminate));

    // worker与task为1:1映射，因此使用workerId作为workId
    const subDurationEvent = task.addSubDurationEvent(workerId.toString()).start();
    subDurationEvent?.setTid(`Worker${workerId}`);

    // 在创建watchWorker时，向主进程的sessionManager发送newWatchWorker的事件
    // 绑定socket与thread、和为socket注册事件，都由sessionManager处理
    if (cluster.isWorker) {
      process.send!({
        type: WatchEvent.NEW_WATCH_WORKER,
        workerId,
      });
    }

    const { promise, resolve, reject } = genPromiseAndTriggers(task, startTime);
    task.pendingPromises.add(promise);

    this.createWorkerOnMessageListener(worker, resolve, reject, task, workerId, callback, subDurationEvent);

    this.createWorkerOnErrorListener(worker, reject, task, workerId);
    return workerId.toString();
  }

  createWorkerOnMessageListener(
    worker: Worker,
    resolve: (value: unknown) => void,
    reject: (reason?: unknown) => void,
    task: CoreTaskImpl,
    workerId: number,
    callback: Function,
    subDurationEvent: DurationEvent,
  ) {
    // 监听来自worker线程的消息，转发给sessionManager或执行相应操作
    worker.on('message', async (response) => {
      if (!response) {
        reject(new Error('Undefined response.'));
        return;
      }
      if (response.event) {
        try {
          await handleResponse(response, task, workerId, true, callback, subDurationEvent);
          resolve('succeed');
        } catch (e) {
          reject(e);
        }
        return;
      }
      if (
        response.type === WatchEvent.WATCH_COMPILE_RESULT ||
        response.type === WatchEvent.WATCH_START ||
        response.type === WatchEvent.WATCH_RESULT ||
        response.type === WatchEvent.WATCH_LOG
      ) {
        log.debug(`watch worker: send response to session manager. Response type: ${response.type}`);
        // 尝试执行用户的hook，如果用户的自定义hook报错，则通过log打印出去，但是不阻断下面的流程
        try {
          await HvigorLifecycleHook.getInstance().runHook(HookType.onWatchWorkerMessage, response, subDurationEvent);
        } catch (e: any) {
          log.error(`custom onWatchWorkerMessage hook exception: ${e.message}`);
        }
        if (cluster.isWorker) {
          response.hotCompile === 'hotCompile' ?
            process.send!({ type: `${response.content === true ? WORKER_PROCESS_EVENT_ID.FINISHED : WORKER_PROCESS_EVENT_ID.FAILED}` }) :
            process.send!({ ...response, workerId });
        }
        return;
      }
      if (response.type === WatchEvent.TERMINATE_WORKER) {
        // 目前现在的代码里，terminateWorker的操作本身是由client侧发起的，但会存在一个给worker线程告知之后的回调事件
        // 所以不需要给sessionManager发送告知该事件
        log.debug('watch worker: worker is ready to be terminated.');
        this.terminateWorker(response.content);
        return;
      }
      reject(new Error(`Unknown watch event: ${response}.`));
    });
  }

  createWorkerOnErrorListener(worker: Worker, reject: (reason?: unknown) => void, task: CoreTaskImpl, workerId: number) {
    worker.on('error', async (error) => {
      log.error(`${task.getName()} watch work[${workerId}] failed.`);
      log.error(error);
      reject(error);
    });
  }

  beforeTerminate(id: number) {
    if (this.workerMap.get(id)?.hasCustomTerminate) {
      this.workerMap.get(id)?.worker.postMessage({
        type: WatchEvent.TERMINATE_WORKER,
        content: id,
      });
    } else {
      this.terminateWorker(id);
    }
  }

  private terminateWorker(id: number): void {
    this.workerMap.get(id)?.worker.terminate();
    this.workerMap.delete(id);
    log.debug(`Server currently has ${this.workerMap.size} ${this.workerMap.size > 1 ? 'watch-workers' : 'watch-worker'}`);
  }

  /**
   * 监听主线程传递的来自socket的信息，传递给对应的worker线程
   * @private
   */
  private addListenersOnSessionManager() {
    if (cluster.isWorker) {
      cluster.worker!.on('message', (response) => {
        if (response.workerId === undefined || response.workerId === 0) {
          log.debug('watch worker: worker id should be larger than 0. Nothing will be sent to any worker thread.');
          return;
        }
        if (response.type === WatchEvent.WATCH_COMPILE_DATA) {
          log.debug('watch worker: receive watch compile tata.');
          this.getWorker(response.workerId)?.postMessage(response.content);
        } else if (response.type === WatchEvent.CLOSE_WATCH) {
          this.beforeTerminate(response.workerId);
        }
      });
    }
  }
}

interface QueueItem {
  workerId: number;
  task: CoreTaskImpl;
  filename: string;
  callback: Function;
  options?: WorkerOptions;
}

interface PromiseItem {
  resolve: (value: unknown) => void;
  reject: (reason?: unknown) => void;
}

/**
 * 普通worker，编译完成后直接terminate。
 * 目前用于预览时级联编译entry所依赖的所有hsp。
 */
class NormalWorker {
  private static runningThreadCnt = 0;
  private static workQueue: QueueItem[] = [];
  private static promiseMap: Map<number, PromiseItem> = new Map<number, PromiseItem>();

  // maxWorkerNum至少为1
  private readonly maxWorkerNum = PoolConstant.MAX_POOL_NUM > 1 ? PoolConstant.MAX_POOL_NUM - 1 : 1;

  createWorker(task: CoreTaskImpl, filename: string, hasCustomTerminate: boolean, callback: Function, options?: WorkerOptions): string {
    const startTime = Number(process.hrtime.bigint());
    const workerId = WorkerCountManager.allocateWorkerId();
    const { promise, resolve, reject } = genPromiseAndTriggers(task, startTime);
    task.pendingPromises.add(promise);
    NormalWorker.promiseMap.set(workerId, { resolve, reject });
    if (NormalWorker.runningThreadCnt >= this.maxWorkerNum) {
      log.debug(`Creating worker failed since runningThreadCnt is not smaller than ${this.maxWorkerNum}.`);
      NormalWorker.workQueue.push({
        workerId,
        task,
        filename,
        callback,
        options,
      });
    } else {
      log.debug(`Creating worker with id ${workerId}.`);
      this.executeWorker(workerId, task, filename, callback, options);
    }
    return workerId.toString();
  }

  reset() {
    NormalWorker.runningThreadCnt = 0;
    NormalWorker.workQueue.length = 0;
    NormalWorker.promiseMap.clear();
  }

  private executeWorker(workerId: number, task: CoreTaskImpl, filename: string, callback: Function, options?: WorkerOptions) {
    const worker: Worker = new Worker(filename, options);
    NormalWorker.runningThreadCnt++;

    // Safety: executeWorker前已将workerId对应的resolve和reject存入map
    const { resolve, reject } = NormalWorker.promiseMap.get(workerId)!;
    NormalWorker.promiseMap.delete(workerId);

    // worker与task为1:1映射，因此使用workerId作为workId
    const subDurationEvent = task.addSubDurationEvent(workerId.toString()).start();
    subDurationEvent?.setTid(`Worker${workerId}`);

    worker.on('message', async (response) => {
      if (!response) {
        reject(new Error('Undefined response.'));
        return;
      }
      if (response.event) {
        try {
          await handleResponse(response, task, workerId, false, callback, subDurationEvent);
          worker.terminate();
          NormalWorker.runningThreadCnt--;
          if (NormalWorker.workQueue.length) {
            // Safety: 上面已判断workerQueue非空
            const workerQueueItem = NormalWorker.workQueue.shift()!;
            log.debug(`Normal worker queue is not empty. Create worker with id: ${workerQueueItem.workerId}.`);
            this.executeWorker(workerQueueItem.workerId, workerQueueItem.task, workerQueueItem.filename, workerQueueItem.callback, workerQueueItem.options);
          }
          resolve(true);
        } catch (e) {
          reject(e);
        }
        return;
      }
      if (
        response.type === WatchEvent.WATCH_COMPILE_RESULT ||
        response.type === WatchEvent.WATCH_START ||
        response.type === WatchEvent.WATCH_RESULT ||
        response.type === WatchEvent.WATCH_LOG
      ) {
        return;
      }
      if (response.type === WatchEvent.TERMINATE_WORKER) {
        // normal worker中不可能收到此事件
        log.debug('Impossible response type in normal worker: TERMINATE_WORKER');
        return;
      }
      reject(new Error(`Unknown watch event: ${response}.`));
    });

    worker.on('error', async (error) => {
      log.debug(`${task.getName()} normal work[${workerId}] failed.`);
      reject(error);
    });
  }
}

export const watchWorker: WatchWorker = new WatchWorker();
export const normalWorker: NormalWorker = new NormalWorker();
